package com.zyb.dfs.backupnode.server;

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;

/**
 * 从namenode同步editslog组件
 */
public class EditsLogFetcher extends Thread {

    public static final Integer BACKUP_NODE_FETCH_SIZE = 10;

    private BackupNode backupNode;

    private NameNodeRpcClient nameNode;

    private FSNamesystem namesystem;

    public EditsLogFetcher(BackupNode backupNode, FSNamesystem namesystem, NameNodeRpcClient nameNode) {
        this.backupNode = backupNode;
        this.nameNode = nameNode;
        this.namesystem = namesystem;
    }

    @Override
    public void run() {
        while (backupNode.isRunning()) {
            try {
                JSONArray editslogs = nameNode.fetchEditsLog(namesystem.getSyncTxid());
                if (editslogs.size() == 0) {
                    Thread.sleep(1000);
                    continue;
                }

                if (editslogs.size() < BACKUP_NODE_FETCH_SIZE) {
                    Thread.sleep(1000);
                }

                for (int i = 0; i < editslogs.size(); i++) {
                    JSONObject editslog = editslogs.getJSONObject(i);
                    String op = editslog.getString("OP");
                    if ("MKDIR".equals(op)) {
                        String path = editslog.getString("PATH");
                        long txid = editslog.getLong("TXID");
                        namesystem.mkdir(txid, path);
                    } else if ("CREATE".equals(op)) {
                        String filename = editslog.getString("PATH");
                        try {
                            namesystem.create(filename);
                        } catch (Exception e) {
                            e.printStackTrace();
                        }
                    }
                }
            } catch (Exception e) {
                e.getStackTrace();
            }
        }
    }
}
